/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.rocketmq.spring.starter.core;

import java.nio.charset.Charset;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;

import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.messaging.MessagingException;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.SendCallback;
import com.aliyun.openservices.ons.api.SendResult;
import com.fasterxml.jackson.databind.ObjectMapper;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class RocketMQTemplate implements InitializingBean, DisposableBean {

    @Getter
    @Setter
    private Producer aliyunProducer;

    @Setter
    @Getter
    private ObjectMapper objectMapper = new ObjectMapper();

    @Getter
    @Setter
    private String charset = "UTF-8";
    
    /**
	 * 环境前缀
	 */
    @Setter
	private String environmentPrefix;

    /**
     * 同步发送消息
     * @param topic 消息主题, 最长不超过255个字符; 由a-z, A-Z, 0-9, 以及中划线"-"和下划线"_"构成.
     * @param tag 消息标签, 请使用合法标识符, 尽量简短且见名知意
     * @param key 业务主键
     * @param payload 消息体, 消息体长度默认不超过4M, 具体请参阅集群部署文档描述.
     * @param userProperties  添加用户自定义属性键值对; 该键值对在消费消费时可被获取.也可用于做SQL属性过滤 
     * @param startDeliverTime  设置消息的定时投递时间（绝对时间),最大延迟时间为7天.
     * </p>
     * <ol>
     *     <li>延迟投递: 延迟3s投递, 设置为: System.currentTimeMillis() + 3000;</li>
     *     <li>定时投递: 2016-02-01 11:30:00投递, 设置为: new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse("2016-02-01 11:30:00").getTime()</li>
     * </ol>
     * @return {@link SendResult}
     * 2018年3月23日 zhaowg
     */
    public SendResult syncSend(String topic,String tag,String keys,Object payload,Map<String, String> userProperties,Long startDeliverTime) {
        if (Objects.isNull(topic) || Objects.isNull(payload)) {
            log.info("同步消息发送失败，主题和消息不能为空");
            throw new IllegalArgumentException("同步消息发送失败，主题和消息不能为空");
        }

        try {
        	long now = System.currentTimeMillis();
        	
        	Message rocketMsg = new Message(environmentPrefix+"_"+topic, tag, keys, convertToRocketMsg(payload));
        	if(userProperties!=null && !userProperties.isEmpty()){
        		for (Entry<String, String> userProp : userProperties.entrySet()) {
        			rocketMsg.putUserProperties(userProp.getKey(), userProp.getValue());
				}
        	}
        	if(startDeliverTime!=null){
        		//设置定时发送时间
        		rocketMsg.setStartDeliverTime(startDeliverTime);
        	}
			//阿里云发送
        	SendResult sendResult = aliyunProducer.send(rocketMsg);
            long costTime = System.currentTimeMillis() - now;
            log.debug("发送消息耗时： {} ms, msgId:{}", costTime, sendResult.getMessageId());
            return sendResult;
        } catch (Exception e) {
            log.info("同步发送失败. topic:{}, message:{} ", topic, payload);
            throw new MessagingException(e.getMessage(), e);
        }
    }

	/**
     * Same to {@link #syncSend(String, String, String, Object, Map, Long)}.
     * @param topic 消息主题, 最长不超过255个字符; 由a-z, A-Z, 0-9, 以及中划线"-"和下划线"_"构成.
     * @param tag 消息标签, 请使用合法标识符, 尽量简短且见名知意
     * @param key 业务主键
     * @param payload 消息体, 消息体长度默认不超过4M, 具体请参阅集群部署文档描述.
     * @return {@link SendResult}
     * 2018年3月23日 zhaowg
     */
    public SendResult syncSend(String topic,String tag,String keys, Object payload) {
        return syncSend(topic, tag, keys, payload, null, null);
    }
    /**
     * Same to {@link #syncSend(String, String, String, Object)}.
     * @param topic 消息主题, 最长不超过255个字符; 由a-z, A-Z, 0-9, 以及中划线"-"和下划线"_"构成.
     * @param tag 消息标签, 请使用合法标识符, 尽量简短且见名知意
     * @param payload 消息体, 消息体长度默认不超过4M, 具体请参阅集群部署文档描述.
     * @return {@link SendResult}
     * 2018年3月23日 zhaowg
     */
    public SendResult syncSend(String topic,String tag, Object payload) {
        return syncSend(topic, tag,null, payload);
    }

    /**
     * 异步发送消息
     * @param topic 消息主题, 最长不超过255个字符; 由a-z, A-Z, 0-9, 以及中划线"-"和下划线"_"构成.
     * @param tag 消息标签, 请使用合法标识符, 尽量简短且见名知意
     * @param key 业务主键
     * @param payload 消息体, 消息体长度默认不超过4M, 具体请参阅集群部署文档描述.
     * @param userProperties  添加用户自定义属性键值对; 该键值对在消费消费时可被获取.也可用于做SQL属性过滤 
     * @param startDeliverTime  设置消息的定时投递时间（绝对时间),最大延迟时间为7天.
     * </p>
     * <ol>
     *     <li>延迟投递: 延迟3s投递, 设置为: System.currentTimeMillis() + 3000;</li>
     *     <li>定时投递: 2016-02-01 11:30:00投递, 设置为: new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse("2016-02-01 11:30:00").getTime()</li>
     * </ol>
     * @param sendCallback 发送完成要执行的回调函数
     * 2018年3月23日 zhaowg
     */
    public void asyncSend(String topic,String tag,String keys,Object payload,Map<String, String> userProperties,
    		Long startDeliverTime,SendCallback sendCallback) {
        if (Objects.isNull(topic) || Objects.isNull(payload)) {
            log.info("异步消息发送失败，主题和消息不能为空");
            throw new IllegalArgumentException("异步消息发送失败，主题和消息不能为空");
        }
        try {
        	long now = System.currentTimeMillis();
        	
        	Message rocketMsg = new Message(environmentPrefix+"_"+topic, tag, keys, convertToRocketMsg(payload));
        	if(userProperties!=null && !userProperties.isEmpty()){
        		for (Entry<String, String> userProp : userProperties.entrySet()) {
        			rocketMsg.putUserProperties(userProp.getKey(), userProp.getValue());
				}
        	}
        	if(startDeliverTime!=null){
        		//设置定时发送时间
        		rocketMsg.setStartDeliverTime(startDeliverTime);
        	}
			//阿里云发送
        	aliyunProducer.sendAsync(rocketMsg, sendCallback);
            long costTime = System.currentTimeMillis() - now;
            log.debug("发送消息耗时： {} ms", costTime);
        } catch (Exception e) {
            log.info("异步发送失败. topic:{}, message:{} ", topic, payload);
            throw new MessagingException(e.getMessage(), e);
        }
    }
	/**
     * Same to {@link #asyncSend(String, String, String, Object, Map, Long, SendCallback)}.
     * @param topic 消息主题, 最长不超过255个字符; 由a-z, A-Z, 0-9, 以及中划线"-"和下划线"_"构成.
     * @param tag 消息标签, 请使用合法标识符, 尽量简短且见名知意
     * @param key 业务主键
     * @param payload 消息体, 消息体长度默认不超过4M, 具体请参阅集群部署文档描述.
     * @param sendCallback 发送完成要执行的回调函数
     * @return {@link SendResult}
     * 2018年3月23日 zhaowg
     */
    public void asyncSend(String topic,String tag,String keys, Object payload,SendCallback sendCallback) {
        asyncSend(topic, tag, keys, payload, null, null,sendCallback);
    }
    /**
     * Same to {@link #asyncSend(String, String, String, Object,SendCallback)}.
     * @param topic 消息主题, 最长不超过255个字符; 由a-z, A-Z, 0-9, 以及中划线"-"和下划线"_"构成.
     * @param tag 消息标签, 请使用合法标识符, 尽量简短且见名知意
     * @param payload 消息体, 消息体长度默认不超过4M, 具体请参阅集群部署文档描述.
     * @param sendCallback 发送完成要执行的回调函数
     * @return {@link SendResult}
     * 2018年3月23日 zhaowg
     */
    public void asyncSend(String topic,String tag, Object payload,SendCallback sendCallback) {
        asyncSend(topic, tag,null, payload,sendCallback);
    }
    /**
     * 服务器不应答，无法保证消息是否成功到达服务器
     * @param topic 消息主题, 最长不超过255个字符; 由a-z, A-Z, 0-9, 以及中划线"-"和下划线"_"构成.
     * @param tag 消息标签, 请使用合法标识符, 尽量简短且见名知意
     * @param key 业务主键
     * @param payload 消息体, 消息体长度默认不超过4M, 具体请参阅集群部署文档描述.
     * @param userProperties  添加用户自定义属性键值对; 该键值对在消费消费时可被获取.也可用于做SQL属性过滤 
     * @param startDeliverTime  设置消息的定时投递时间（绝对时间),最大延迟时间为7天.
     * </p>
     * <ol>
     *     <li>延迟投递: 延迟3s投递, 设置为: System.currentTimeMillis() + 3000;</li>
     *     <li>定时投递: 2016-02-01 11:30:00投递, 设置为: new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse("2016-02-01 11:30:00").getTime()</li>
     * </ol>
     * 2018年3月23日 zhaowg
     */
    public void sendOneWay(String topic,String tag,String keys,Object payload,Map<String, String> userProperties,
    		Long startDeliverTime) {
        if (Objects.isNull(topic) || Objects.isNull(payload)) {
            log.info("sendOneWay消息发送失败，主题和消息不能为空");
            throw new IllegalArgumentException("sendOneWay消息发送失败，主题和消息不能为空");
        }
        try {
        	long now = System.currentTimeMillis();
        	
        	Message rocketMsg = new Message(environmentPrefix+"_"+topic, tag, keys, convertToRocketMsg(payload));
        	if(userProperties!=null && !userProperties.isEmpty()){
        		for (Entry<String, String> userProp : userProperties.entrySet()) {
        			rocketMsg.putUserProperties(userProp.getKey(), userProp.getValue());
				}
        	}
        	if(startDeliverTime!=null){
        		//设置定时发送时间
        		rocketMsg.setStartDeliverTime(startDeliverTime);
        	}
			//阿里云发送
        	aliyunProducer.sendOneway(rocketMsg);
            long costTime = System.currentTimeMillis() - now;
            log.debug("发送消息耗时： {} ms", costTime);
        } catch (Exception e) {
            log.info("sendOneWay发送失败. topic:{}, message:{} ", topic, payload);
            throw new MessagingException(e.getMessage(), e);
        }
    }
	/**
     * Same to {@link #sendOneWay(String, String, String, Object, Map, Long)}.
     * @param topic 消息主题, 最长不超过255个字符; 由a-z, A-Z, 0-9, 以及中划线"-"和下划线"_"构成.
     * @param tag 消息标签, 请使用合法标识符, 尽量简短且见名知意
     * @param key 业务主键
     * @param payload 消息体, 消息体长度默认不超过4M, 具体请参阅集群部署文档描述.
     * 2018年3月23日 zhaowg
     */
    public void sendOneWay(String topic,String tag,String keys, Object payload) {
    	sendOneWay(topic, tag, keys, payload, null, null);
    }
    /**
     * Same to {@link #sendOneWay(String, String, String, Object)}.
     * @param topic 消息主题, 最长不超过255个字符; 由a-z, A-Z, 0-9, 以及中划线"-"和下划线"_"构成.
     * @param tag 消息标签, 请使用合法标识符, 尽量简短且见名知意
     * @param payload 消息体, 消息体长度默认不超过4M, 具体请参阅集群部署文档描述.
     * 2018年3月23日 zhaowg
     */
    public void sendOneWay(String topic,String tag, Object payload) {
    	sendOneWay(topic, tag,null, payload);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
    	if(aliyunProducer != null){
    		log.info("开始启动阿里云[环境标识："+environmentPrefix+"]生产者");
    		aliyunProducer.start();
    	}
    }

    /**
     * 转换对象为字节
     * @param msgObj
     * @return
     * 2018年3月23日 zhaowg
     */
    private byte[] convertToRocketMsg(Object msgObj) {
        byte[] payloads;

        if (msgObj instanceof String) {
            payloads = ((String) msgObj).getBytes(Charset.forName(charset));
        } else {
            try {
                String jsonObj = this.objectMapper.writeValueAsString(msgObj);
                payloads = jsonObj.getBytes(Charset.forName(charset));
            } catch (Exception e) {
                throw new RuntimeException("convert to RocketMQ message failed.", e);
            }
        }
        return payloads;
    }


    @Override
    public void destroy() {
        if(Objects.nonNull(aliyunProducer)){
        	log.info("开始关闭阿里云[环境标识："+environmentPrefix+"]生产者");
        	aliyunProducer.shutdown();
        }
    }
}
